/**  
  -- The Performance Driven Lime Project -- 

  A self adaptive tuple space for mobile agents and hosts by dynamic
  adapting tuple's replication policy.
  
  Copyright (C) 2008
  Xing Jiankuan, Pan Wenhui, Huang Xiaodong, Liang Qian
  SAGroup of Tsinghua University

  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation; either
  version 2.1 of the License, or (at your option) any later version.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  Lesser General Public License for more details.

  You should have received this copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 
 */

package sagroup.dracots;

import java.util.ArrayList;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.LinkedList;
import java.util.List;
import java.util.Timer;
import java.util.logging.Logger;

import lights.adapters.Field;
import lights.adapters.Tuple;
import lights.interfaces.ITuple;
import lights.interfaces.IValuedField;
import lime2.CommunicationMgr;
import lime2.InvalidLocationException;
import lime2.LimeConstants;
import lime2.LimeServer;
import lime2.LimeServerID;
import lime2.LimeSystemTupleSpace;
import lime2.LimeTuple;
import lime2.LocalOpMgr;
import lime2.Location;
import lime2.Policies;
import lime2.PolicyException;
import lime2.Reaction;
import lime2.ReactionID;
import lime2.ReactionOp;
import lime2.Reconciliator;
import lime2.Result;
import lime2.Translator;
import lime2.TupleID;
import lime2.TupleOp;
import sagroup.dracots.commands.AddReplicaRequestByEntryCommand;
import sagroup.dracots.commands.ClearReplicaCommand;
import sagroup.dracots.commands.PushCommand;
import sagroup.dracots.commands.RemoveReplicaRequestCommand;
import sagroup.dracots.commands.UpdateReplicaCommand;
import sagroup.dracots.policy.PolicyReaction;
import sagroup.dracots.policy.PolicyRegistrar;
import sagroup.dracots.policy.RPReconciliation;
import sagroup.dracots.policy.RPTuple;
import sagroup.dracots.policy.ReplicationPolicy;
import sagroup.dracots.SystemTupleType;
import sagroup.dracots.TupleType;
import sagroup.util.Util;

/**
 * <P>
 * <b>DracoTS</b> stands for Decentralized Replicable Antoadaptive Co-work Tuple
 * Space.
 * </P>
 * 
 * @author Huang Xiaodong
 * @author Xing Jiankuan
 */
public class DracoTS {
	private ReplicaRequestRegistrar replicaRegistrar;
	private PolicyRegistrar policyRegistrar;
	private CostMonitor monitor;
	private boolean enableMonitor = false;
	private String name; // tuple space name
	private byte policy; // local tuple space policy
	private LocalOpMgr lop; // local operation manager
	LocalReplicaRegistrar lrr = new LocalReplicaRegistrar(
			DracoConstants.DEFAULT_REPLICA_MAX_CAPACITY, this);
	PolicySwitcher switcher;
	/**
	 * this hashtable[String,DracoTupleSpace] keeps track of the existing
	 * DracoTupleSpaces for this host
	 */
	protected static Hashtable DracoTSTable = new Hashtable();

	/**
	 * the host where this tuple space locates
	 */
	protected static LimeServerID localLimeServerID = LimeServer.getServer()
			.getServerID();

	/**
	 * this (long) global counter is used to enforce uniqueness of the numeric
	 * identifiers for the Tuple's version layer IDs
	 */
	private static long NEXT_VID_NUM = -1;

	/**
	 * this (long) global counter is used to enforce uniqueness of the numeric
	 * identifiers for the Tuple's lime layer IDs
	 */
	private static long NEXT_TID_NUM = -1;

	static {
		if(!LogManager.isInitalized())
			LogManager.init();
	}

	/**
	 * Decision Maker
	 */
	private PolicyDecisionMaker dmaker;

	private Logger LOG = Logger.getLogger(this.getClass().getName());

	/**
	 * Constructor. User should use <code>get</code> method to get the reference
	 * to a DracoTS
	 * 
	 * @param name
	 *            DracoTS's name
	 * @param policy
	 *            reconciliation policy
	 * @throws PolicyException
	 */
	protected DracoTS(String name, byte policy) throws PolicyException {
		// 1.tuple space initialization //
		// 
		// registers the name
		this.name = name;
		// registers the policy
		this.policy = policy;
		// creates the LocalOpMgr
		this.lop = new LocalOpMgr(name, policy);
		// starts the LocalOpMgr thread
		this.lop.start();
		// if this host is engaged then new tuplespaces must be shared
		if (LimeServer.getServer().isEngaged())
			CommunicationMgr.get().getCA().joinGroup(name);

		// 2.replication mechanism initialization //
		// 
		monitor = CostMonitor.getMonitor(name);
		replicaRegistrar = new ReplicaRequestRegistrar();
		policyRegistrar = new PolicyRegistrar();

		// add replication policy's replication policy (RP2)
		this.out(new RPTuple(SystemTupleType.REPLICATION_POLICY,
				ReplicationPolicy.FR));

		// add a custom reconciliation job for RPTuples
		Reconciliator.AddReconciliationJob(new RPReconciliation());
		
		// 3.add replication policy switcher
		switcher = new PolicySwitcher(this);

		// 4.self-adaption mechanism initialization //
		// 
		dmaker = PolicyDecisionMaker.getPolicyDecisionMaker(this);

		Timer timer = new Timer();

		timer.schedule(dmaker, DracoConstants.ADAPTATION_PERIOD,
				DracoConstants.ADAPTATION_PERIOD);
	}

	/**
	 * enable or disenable the monitor for log
	 * 
	 * @param enable
	 *            to use monitor to keep a log
	 * @author Huang Xiaodong
	 */
	synchronized public void setMonitorStatus(boolean enable) {
		enableMonitor = enable;
	}

	/**
	 * Returns the next TupleID for this host. <br>
	 * The numeric identifier of the first TupleID created on this host is zero.
	 * Synchronization is necessary since multiple agents (threads) could try to
	 * access the global counter ("tids") when we are in the middle of a
	 * previous invocation of this method (.. this would ruin uniqueness of
	 * tuple identifiers).
	 * 
	 * @return a valid version layered TupleID
	 */
	public synchronized static TupleID nextVID() {
		NEXT_VID_NUM++;
		return new TupleID(Location.HERE.getID(), NEXT_VID_NUM);
	}

	/**
	 * Returns the next TupleID for this host. <br>
	 * The numeric identifier of the first TupleID created on this host is zero.
	 * Synchronization is necessary since multiple agents (threads) could try to
	 * access the global counter ("tids") when we are in the middle of a
	 * previous invocation of this method (.. this would ruin uniqueness of
	 * tuple identifiers).
	 * 
	 * @return valid TupleID
	 */
	public synchronized static TupleID nextTID() {
		NEXT_TID_NUM++;
		return new TupleID(Location.HERE.getID(), NEXT_TID_NUM);
	}

	/**
	 * If a tuple space have existed, return it. Otherwise, create a new one
	 */
	synchronized public static DracoTS get(String name) {
		try {
			return get(name, Policies.DEFAULT);
		} catch (PolicyException e) {
			//impossible to execute here
			return null;
		}
	}

	/**
	 * If a tuple space have existed, return it. Otherwise, create a new one
	 */
	synchronized public static DracoTS get(String name, byte pol)
			throws PolicyException {
		DracoTS ret = null;
		if (name != null) {
			try {
				ret = (DracoTS) DracoTSTable.get(name);
			} catch (ClassCastException e) {
				ret = null;
			}
			if (ret == null) {
				ret = new DracoTS(name, pol);
				// add reactions to handle system commands and policies
				ret.addStrongReaction(new DracoReaction[] {
						new CommandReaction(Reaction.ONCEPERTUPLE, ret),
						new PolicyReaction(Reaction.ONCEPERTUPLE, ret.switcher) });
				DracoTSTable.put(name, ret);
			} else if (pol != ret.getLimePolicy())
				throw new PolicyException("policy has already been set to "
						+ Translator.translatePolicy(ret.getLimePolicy()));
		}
		return ret;
	}

	/**
	 * Returns the policy for the DracoTupleSpace
	 * 
	 * @return policy
	 */
	public byte getLimePolicy() {
		return this.policy;
	}

	public ReplicaRequestRegistrar getReplicaRequestRegistrar() {
		return replicaRegistrar;
	}

	public PolicyRegistrar getPolicyRegistrar() {
		return policyRegistrar;
	}

	/**
	 * Returns the name of this DracoTS.
	 * 
	 * @return name
	 */
	public String getName() {
		return this.name;
	}

	/**
	 * Returns the names of all the TVRLimeTupleSpaces for this host.
	 * 
	 * @return array of names
	 */
	public static String[] getNames() {
		String[] ret = new String[DracoTSTable.size()];
		int i = 0;
		for (Enumeration e = DracoTSTable.keys(); e.hasMoreElements();)
			ret[i++] = (String) e.nextElement();
		return ret;
	}

	/**
	 * Reads a tuple matching a specific TVRLimeTemplate.
	 * 
	 * @param template
	 *            A DracoTS template
	 * @return result DracoTuple
	 * 
	 * @author Huang Xiaodong
	 */
	public DracoTuple rd(DracoTuple template) {
		DracoTuple ret = null;
		if (enableMonitor) {
			CostInfo info = startMonitorOp(LimeConstants.RD);
			ret = doRd(template);
			endMonitorOp(info, ret, false);
		} else {
			ret = doRd(template);
		}
		
		this.recordReadCount(ret);
		return ret;
	}

	private CostInfo startMonitorOp(byte op) {
		CostInfo info = new CostInfo();
		info.operationType = op;
		info.startTime = System.nanoTime();
		return info;
	}

	private void endMonitorOp(CostInfo info, DracoTuple ret, boolean isWrite) {
		info.endTime = System.nanoTime();
		info.processTime = info.endTime - info.startTime;
		if (isWrite)
			info.totalWriteTime = info.processTime;
		else
			info.totalReadTime = info.processTime;

		if (ret != null) {
			info.type = ret.getType();
			info.tupleID = ret.getVID();
			info.sourceLocation = ret.getCurr();
			info.sourceLocation = ret.getDest();
			info.size = ret.getLimeWrapper().tuple.length();
		}
		monitor.recordOperation(info);
	}

	private void endMonitorOp(CostInfo info, DracoTuple[] ret, boolean isWrite) {
		info.endTime = System.nanoTime();
		info.processTime = info.endTime - info.startTime;
		if (isWrite)
			info.totalWriteTime = info.processTime;
		else
			info.totalReadTime = info.processTime;

		if (ret != null && ret.length > 0) {
			info.type = ret[0].getType();
			info.tupleID = ret[0].getVID();
			info.sourceLocation = ret[0].getCurr();
			info.sourceLocation = ret[0].getDest();
			for (int i = 0; i < ret.length; i++) {
				info.size += ret[i].getLimeWrapper().tuple.length();
			}
		}
		monitor.recordOperation(info);
	}

	/**
	 * Perform the actual rd operation
	 * 
	 * @param template
	 * @return
	 * @author Xing Jiankuan @
	 */
	DracoTuple doRd(DracoTuple template) {
		DracoTuple ret = null;
		// try to read local replica if template indicate remote host or
		// unspecified
		// if it is FR, always replica can be found
		if (template instanceof ShareTuple && !template.getCurr().equals(Location.HERE)) {

			ShareTuple stemplate = (ShareTuple) template
			.duplicateTemplate(template.getDest());

			stemplate.setCurr(Location.HERE);
			stemplate.setDest(Location.HERE);
			stemplate.setReplica(true);
			ret = lrr.rrdp(stemplate);
			if (ret != null) { // found
				assert (ret instanceof ShareTuple);
				lrr.touch((ShareTuple) ret); // renew replica position in LRU
                System.out.print(ret.vid+" is a replica.");
				return ret;
			} else {
//				// not found, reset original template's curr and set
//				template.setCurr(curr);
//				template.setDest(dest);
			}
		}
		// perform the conventional rd
		ret = performUniOp(LimeConstants.RD, template);

		// Here to check the policy.
		if (ret != null && ret instanceof ShareTuple) {
			ShareTuple st = (ShareTuple) ret;
			
			new Thread(new TryRC(st)).start();
		} 

		return ret;
	}

	
	private void recordReadCount(DracoTuple dt) {
	

		if (dt != null && dt instanceof ShareTuple) {
			ShareTuple st = (ShareTuple) dt;
			if ((st.isReplica && !st.getMasterCurr().isHere())
					|| isRemoteMaster(st)) {
				TupleTypeInfo.getTypeRecord(name, st.getType())
						.incrReadCount(1);
				dmaker.insertReadRecord(st);
			}
		}

	}

	// public void updateVersionTable(DracoTuple t) {
	// // if there is t's replica at local, no need to calculate awc.
	// if (lrr.containsVID(t.getVID()))
	// return;
	//
	// if (!t.isLocal()) {
	//
	// ShareTuple st = (ShareTuple) t;
	// dmaker.updateVersionRecord(st.getVID(), st.getType(), st.getVer());
	//
	// }
	// }

	/**
	 * Non-blocking read of a tuple matching a specific template.
	 * 
	 * @param template
	 *            template
	 * @return result a copy of draco tuple @
	 */
	public DracoTuple rdp(DracoTuple template) {
		DracoTuple ret = null;
		if (enableMonitor) {
			CostInfo info = startMonitorOp(LimeConstants.RDP);
			ret = doRdp(template);
			endMonitorOp(info, ret, false);
		} else {
			ret = doRdp(template);
		}

		this.recordReadCount(ret);
		return ret;
	}

	/**
	 * do the actual rdp operation
	 * 
	 * @param tempalte
	 * @author Xing Jiankuan @
	 */
	DracoTuple doRdp(DracoTuple template) {
		DracoTuple ret = null;
		// try to read local replica if template indicate remote host or
		// unspecified
		// if it is FR, always replica can be found
		if (!template.getCurr().equals(Location.HERE)
				&& template instanceof ShareTuple) {

			ShareTuple stemplate = (ShareTuple) template
					.duplicateTemplate(template.getDest());

			stemplate.setCurr(Location.HERE);
			stemplate.setDest(Location.HERE);
			stemplate.setReplica(true);
			ret = lrr.rrdp(stemplate);
			if (ret != null) { // found
				assert (ret instanceof ShareTuple);
				lrr.touch((ShareTuple) ret); // renew replica position in LRU
//				System.out.print(ret.vid+" is a replica.");
				return ret;

			} else {
				// not found, reset original template's curr and set
				// template.setCurr(curr);
				// template.setDest(dest);
			}
		}
		// perform the conventional rdp
		ret = performUniOp(LimeConstants.RDP, template);

		// Here to check the policy.
		if (ret != null && ret instanceof ShareTuple) {
			ShareTuple st = (ShareTuple) ret;
			new Thread(new TryRC(st)).start();
		}

		return ret;
	}

	/**
	 * if t is master, and t is remote, return true; otherwise, t's master is
	 * remote, return true.
	 * 
	 * @param t
	 * @return
	 * @author Xing Jiankuan
	 */
	private boolean isRemoteMaster(DracoTuple dt) {
		if (dt instanceof ShareTuple) {
			ShareTuple t = (ShareTuple) dt;
			if (t.isMaster())
				return !t.limeWrapper.cur.isHere();
			else
				return !t.masterCurr.isHere();
		} else {
			return false;
		}
	}

	/**
	 * Try Read and Cache.
	 * 
	 * @param st
	 * @author Huang Xiaodong @
	 */
	private void tryRC(ShareTuple st) {
		TupleType targetType = st.getType();
		// If targetType is SystemTupleType.POLICY, no need to look up
		// policyRegistrar, it will be treated as FR, and processed in
		// PolicyReactionListener. Only add cache for remote tuples.
		if (!targetType.equals(SystemTupleType.REPLICATION_POLICY)
				&& !st.isLocal()) {
			ReplicationPolicy policy = policyRegistrar.getReplicaPolicy(
					targetType, Location.HERE);
			// If the replication policy is RC, and LocalHost haven't got a
			// replication, then send an AddReplicaRequestCommand after
			// replicate the tuple
			if (policy != null && (policy.equals(ReplicationPolicy.RC))) {
				ShareTuple vidTemplate = ShareTuple.createVIDTemplate(st
						.getVID());
				vidTemplate.setCurr(Location.HERE);
				if (lrr.rrdp(vidTemplate) == null) {
					ShareTuple replica = null;
					try {
						replica = st.getClass().newInstance();
					} catch (InstantiationException e) {
						e.printStackTrace();
					} catch (IllegalAccessException e) {
						e.printStackTrace();
					}
					replica = st.replicate(Location.HERE);
					lrr.addReplicaAndTouch(replica); // add replica to local
					// host and touch
					// immediately

					// add replica request
					AddReplicaRequestByEntryCommand.ReplicaRequestEntry entry = new AddReplicaRequestByEntryCommand.ReplicaRequestEntry(
							st.getVID(), st.getVer());
					addReplica(
							st.getCurr(),
							new AddReplicaRequestByEntryCommand.ReplicaRequestEntry[] { entry });
				}
			}
		}
	}

	/**
	 * Try Read and Cache for a array.
	 * 
	 * @param st
	 * @author Huang Xiaodong @
	 */
	private void tryRCs(DracoTuple[] dt) {
		ShareTuple[] replicas = new ShareTuple[dt.length];
		
		HashMap<Location, LinkedList<AddReplicaRequestByEntryCommand.ReplicaRequestEntry>> replicaRequests =
			new HashMap<Location, LinkedList<AddReplicaRequestByEntryCommand.ReplicaRequestEntry>> ();
		
		for (int i = 0; i < dt.length; i++) {
			if (dt[i] == null || !(dt[i] instanceof ShareTuple)) {
				replicas[i] = null;
			} else {
				ShareTuple st = ((ShareTuple) dt[i]);
				TupleType targetType = st.getType();
				// If targetType is SystemTupleType.POLICY, no need to look up
				// policyRegistrar, it will be treated as FR, and processed in
				// PolicyReactionListener. Only add cache for remote tuples.
				if (!targetType.equals(SystemTupleType.REPLICATION_POLICY)
						&& !st.isLocal()) {
					ReplicationPolicy policy = policyRegistrar
							.getReplicaPolicy(targetType, Location.HERE);
					// If the replication policy is RC, and LocalHost haven't
					// got a
					// replication, then send an AddReplicaRequestCommand after
					// replicate the tuple
					if (policy != null && (policy.equals(ReplicationPolicy.RC))) {
						ShareTuple vidTemplate = ShareTuple
								.createVIDTemplate(st.getVID());
						vidTemplate.setCurr(Location.HERE);
						if (doRdp(vidTemplate) == null) {
							ShareTuple replica = null;
							try {
								replica = st.getClass().newInstance();
							} catch (InstantiationException e) {
								e.printStackTrace();
							} catch (IllegalAccessException e) {
								e.printStackTrace();
							}
							replica = st.replicate(Location.HERE);
							replicas[i] = replica;
							// lrr.addReplicaAndTouch(replica); // add replica
							// to local host and touch immediately

							// add replica request
							AddReplicaRequestByEntryCommand.ReplicaRequestEntry entry = new AddReplicaRequestByEntryCommand.ReplicaRequestEntry(
									st.getVID(), st.getVer());
							LinkedList<AddReplicaRequestByEntryCommand.ReplicaRequestEntry> replicaRequestList;
							if(!replicaRequests.containsKey(st.getCurr())){
								replicaRequestList = new LinkedList<AddReplicaRequestByEntryCommand.ReplicaRequestEntry>();
								replicaRequests.put(st.getCurr(), replicaRequestList);
							}
							else{
								replicaRequestList = replicaRequests.get(st.getCurr());
							}
							replicaRequestList.add(entry);
							
//							addReplica(
//									st.getCurr(),
//									new AddReplicaRequestByEntryCommand.ReplicaRequestEntry[] { entry });
						}
					} else
						replicas[i] = null;
				} else
					replicas[i] = null;
			}
		}
		// Out to local first
		lrr.addReplicasAndTouch(replicas);
		// Add the replica request then
		for(Location replicaLoc: replicaRequests.keySet()){
			AddReplicaRequestByEntryCommand.ReplicaRequestEntry[] replicaArray = 
				new	AddReplicaRequestByEntryCommand.ReplicaRequestEntry[replicaRequests.get(replicaLoc).size()];
			replicaRequests.get(replicaLoc).toArray(replicaArray);
			addReplica(replicaLoc, replicaArray);
		}
	}
	
	/**
	 * Reads tuples matching a specific DracoTuple Template.
	 * 
	 * @param template
	 *            A DracoTuple template
	 * @return result LimeTuple
	 * 
	 * @author Huang Xiaodong
	 * 
	 */
	public DracoTuple[] rdg(DracoTuple template) {

		DracoTuple ret[] = null;
		if (enableMonitor) {
			CostInfo info = startMonitorOp(LimeConstants.RDG);
			ret = performMultiOp(LimeConstants.RDG, template);
			endMonitorOp(info, ret, false);
		} else {
			ret = performMultiOp(LimeConstants.RDG, template);
		}

		
			for(DracoTuple dt:ret)
			  recordReadCount(dt);
		

		return ret;
	}

	/**
	 * Takes a tuple matching a specific DracoTuple Template, the tuple to take
	 * must be a master.
	 * 
	 * @param t
	 *            DracoTuple Template
	 * @return result DracoTuple that is
	 */
	public DracoTuple in(DracoTuple template) {
		DracoTuple ret = null;
		if (template != null) {
			if (template instanceof ShareTuple
					&& !((ShareTuple) template).isMaster())
				throw new IllegalArgumentException(
						"The tuple to take must be a master.");
			else {
				ret = inAny(template);

				// if the tuple removed is a master ShareTuple,
				// need to clear replications
				if (ret instanceof ShareTuple) {
					ShareTuple st = (ShareTuple) ret;
					TupleID vid = st.getVID();
					Location[] wheretos = replicaRegistrar
							.deregisterReplica(vid);
					if (wheretos != null && wheretos.length != 0) {
						ClearReplicaCommand crc = new ClearReplicaCommand(new TupleID[] { vid }, Location.UNSPECIFIED);
						outToMulti(crc, wheretos);
					}
				}
			}
		}
		return ret;
	}

	/**
	 * Takes a tuple matching a specific DracoTuple Template, the tuple can be
	 * either a master or a replication.
	 * 
	 * @param t
	 *            DracoTuple template
	 * @return result LimeTuple
	 */
	private DracoTuple inAny(DracoTuple template) {
		DracoTuple ret = null;
		if (enableMonitor) {
			CostInfo info = startMonitorOp(LimeConstants.IN);
			ret = performUniOp(LimeConstants.IN, template);
			endMonitorOp(info, ret, false);
		} else {
			ret = performUniOp(LimeConstants.IN, template);
		}

		return ret;
	}

	/**
	 * Non-blocking take of a tuple matching a specific DracoTuple template. If
	 * the tuple to take is a ShareTuple, then it must be a master.
	 * 
	 * @param template
	 * @return
	 * @author Huang Xiaodong
	 */
	public DracoTuple inp(DracoTuple template) {
		DracoTuple ret = null;
		if (template != null) {
			if (template instanceof ShareTuple
					&& !((ShareTuple) template).isMaster())
				throw new IllegalArgumentException(
						"The tuple to take must be a master.");
			else {
				ret = inpAny(template);

				// if the tuple removed is a master ShareTuple,
				// need to clear replications
				if (ret != null && ret instanceof ShareTuple) {
					ShareTuple st = (ShareTuple) ret;
					TupleID vid = st.getVID();
					Location[] wheretos = replicaRegistrar
							.deregisterReplica(vid);
					if (wheretos != null && wheretos.length != 0) {
						ClearReplicaCommand crc = new ClearReplicaCommand(new TupleID[] { vid }, Location.UNSPECIFIED);
						outToMulti(crc, wheretos);
					}
				}
			}
		}
		return ret;
	}

	/**
	 * Non-blocking take of a tuple matching a specific DracoTuple template.
	 * 
	 * @param t
	 *            Draco template
	 * @return result Draco tuple
	 */
	DracoTuple inpAny(DracoTuple template) {
		DracoTuple ret = null;
		if (enableMonitor) {
			CostInfo info = startMonitorOp(LimeConstants.INP);
			ret = performUniOp(LimeConstants.INP, template);
			endMonitorOp(info, ret, false);
		} else {
			ret = performUniOp(LimeConstants.INP, template);
		}

		return ret;
	}

	/**
	 * take of tuples matching a specific DracoTuple template. if take of
	 * ShareTuples, they must be masters.
	 * 
	 * @param t
	 *            Draco template
	 * @return result Draco tuples
	 */
	public DracoTuple[] ing(DracoTuple template) {
		DracoTuple[] ret = null;
		if (template != null) {
			if (template instanceof ShareTuple
					&& !((ShareTuple) template).isMaster())
				throw new IllegalArgumentException(
						"The tuples to take must be a master.");
			else {
				ret = ingAny(template);

				// since the tuple removed is master ShareTuple, need to
				// clear replications
				
				//Organize the tuples by their tuple types
				HashMap<TupleType, LinkedList<ShareTuple>> tuples = new HashMap<TupleType, LinkedList<ShareTuple>>();
				for (DracoTuple dt : ret) {
					if(dt instanceof ShareTuple){
						ShareTuple st = (ShareTuple) dt;
						if (st.getDest().isHere() && st.isMaster()) {
							if (!tuples.containsKey(st.type))
								tuples.put(st.type, new LinkedList<ShareTuple>());
							tuples.get(st.type).add(st);
						}	
					}
				}
				
				//filter out tuples that are unnecessary to remove
				ArrayList<TupleID> vids = new ArrayList<TupleID>();
				ArrayList<TupleType> toBeRemoved = new ArrayList<TupleType>();
				for (TupleType type : tuples.keySet()) {
					LinkedList<ShareTuple> sts = tuples.get(type);
					Location[] wheretos = replicaRegistrar.deregisterReplica(sts.get(0).getVID());
					if (wheretos.length == 0) {
						toBeRemoved.add(type);
						continue;
					}

					//for those that need to clear, clear them
					vids.clear();
					for (ShareTuple st : sts) {
						vids.add(st.getVID());
					}
					
					// convert array list to array
					TupleID[] vidArr = new TupleID[vids.size()];
					for (int i = 0; i < vids.size(); i++) {
						vidArr[i] = (TupleID) vids.get(i);
					}
					
					if (vidArr.length == 0)
						return ret;
				
					if(LimeConstants.TREE_MULTICAST) {
						ClearReplicaCommand crc = new ClearReplicaCommand(vidArr, Location.UNSPECIFIED); 
						outToMulti(crc, wheretos);
					} else {
						for(Location target: wheretos) {
							out(new ClearReplicaCommand(vidArr, target));
						}
					}
				}
				
				//clear the unnecessary tuple type
				for(TupleType type: toBeRemoved) {
					tuples.remove(type);
				}
			}
		}
		return ret;
	}

	/**
	 * take of tuples matching a specific DracoTuple template. it takes of
	 * ShareTuples both masters and replications.
	 * 
	 * @param t
	 *            Draco template
	 * @return result Draco tuples
	 */
	public DracoTuple[] ingAny(DracoTuple template) {
		DracoTuple[] ret = null;
		if (enableMonitor) {
			CostInfo info = startMonitorOp(LimeConstants.ING);
			ret = performMultiOp(LimeConstants.ING, template);
			endMonitorOp(info, ret, false);
		} else {
			ret = performMultiOp(LimeConstants.ING, template);
		}
		return ret;
	}

	/**
	 * Update a local Share tuple's content and version. The vid and dest of
	 * tuple <code>newTuple</code> is used to find the original tuple. If found,
	 * the new version tuple is updated to the tuple space; otherwise, nothing
	 * happens, just return a null. All the meta fields in <code>newTuple</code>
	 * is ignored and replaced with meta fields of the replaced tuple, except
	 * the version is incremented.
	 * 
	 * @param newTuple
	 *            the new version tuple object
	 * @return the result of update tuple. If the tuple with vid of t is not
	 *         found, return null
	 * 
	 * @exception IllegalArgumentException
	 *                In one of the following situations will throw
	 *                IllegalArgumentException:<br>
	 *                <ul>
	 *                <li>the input tuple is null</li>
	 *                <li>the input tuple is not a master</li>
	 *                <li>the vid of newTuple represents not a ShareTuple</li>
	 *                <li>the vid of newTuple represents a tuple which is not a
	 *                master</li>
	 *                </ul>
	 * 
	 * @author Zheng Xiang
	 */

	public ShareTuple update(ShareTuple newTuple) {
		// parameter checking
		if (newTuple == null)
			throw new NullPointerException("The new tuple for updating is null");
		if (!newTuple.isMaster())
			throw new IllegalArgumentException("must be a master");

		if (newTuple.getDest().isUnspecified()) {
			throw new IllegalArgumentException(
					"Unspecified is not allowed for update");
		}
		ShareTuple template = ShareTuple.createVIDTemplate(newTuple.getVID());
		template.setDest(newTuple.getDest());
		return doUpdate(template, newTuple);
	}

	/**
	 * Update a local Share tuple's content and version. The
	 * <code>template</code> is used to find the original tuple. If found, the
	 * new version tuple is updated to the tuple space; otherwise, nothing
	 * happens, just return a null. All the meta fields in <code>newTuple</code>
	 * is ignored and replaced with meta fields of the replaced tuple, except
	 * the version is incremented.
	 * 
	 * @param template
	 *            the template to find the tuple
	 * @param newTuple
	 *            the new version tuple object
	 * @return the result of update tuple. If the tuple with vid of t is not
	 *         found, return null
	 * 
	 * @exception IllegalArgumentException
	 *                In one of the following situations will throw
	 *                IllegalArgumentException:<br>
	 *                <ul>
	 *                <li>the input tuple or template is null</li>
	 *                <li>the input tuple is not a master</li>
	 *                <li>the vid of newTuple represents not a ShareTuple</li>
	 *                <li>the vid of newTuple represents a tuple which is not a
	 *                master</li>
	 *                </ul>
	 * 
	 * @author Zheng Xiang
	 */
	public ShareTuple update(ShareTuple template, ShareTuple newTuple) {
		// parameter checking
		if (newTuple == null || newTuple == null)
			throw new IllegalArgumentException(
					"The new tuple for updating is null");
		if (!template.isMaster())
			throw new IllegalArgumentException("must be a master");

		if (template.getDest().isUnspecified()) {
			throw new IllegalArgumentException(
					"Unspecified is not allowed for update");
		}

		return doUpdate(template, newTuple);
	}

	ShareTuple doUpdate(ShareTuple template, ShareTuple newTuple) {
		Result result;
		TupleOp to;

		if (template.getDest().isHere()) {
			// first read out
			DracoTuple rdt = this.doUpdateRdp(template);

			ShareTuple oldt;

			if (rdt == null) // not found
				return null;

			if (!(rdt instanceof ShareTuple))
				throw new IllegalArgumentException(
						"The tuple to be updated isn't a share tuple");
			else
				oldt = (ShareTuple) rdt;

			if (!oldt.isMaster())
				throw new IllegalArgumentException(
						"The tuple to be updated is not a master!");

			oldt.transferMetaData(newTuple);
			newTuple.ver = new Long(newTuple.ver.longValue() + 1L);

			result = new Result();
			to = new TupleOp(LimeConstants.UPDATE, new ITuple[] {
					template.toLimeWrapper().limifyWrapper(),
					newTuple.toLimeWrapper().limifyWrapper() }, result);

		} else {
			result = new Result();
			to = new TupleOp(LimeConstants.UPDATE, new ITuple[] {
					template.toLimeWrapper().limifyWrapper(),
					newTuple.toLimeWrapper().limifyWrapper() }, result);

		}

		lop.postOperation(to);

		ShareTuple ret = null;
		try {
			ret = (ShareTuple) DracoTuple.create((ITuple) result.get());
		} catch (ClassNotFoundException e) {
			throw new RuntimeException(e);
		}
		// Here to update replications.
		if (ret != null) {
			Location[] wheretos = replicaRegistrar.updateReplica(ret.getVID(),
					ret.getVer());
			if (wheretos != null) {
				UpdateReplicaCommand urc = new UpdateReplicaCommand(new ShareTuple[] { ret.replicate(null)}, Location.UNSPECIFIED);
				outToMulti(urc, wheretos);
			}
		}

		return ret;
	}

	/**
	 * Out a Draco tuple directly
	 * 
	 * @param t the draco tuple that have been created
	 * @return the out result tuple
	 * @author Xing Jiankuan
	 */
	public DracoTuple out(DracoTuple t) {
		DracoTuple ret = null;
		if (t instanceof ShareTuple && ((ShareTuple) t).isReplica)
			throw new IllegalArgumentException("Cannot out a replica");

		if (enableMonitor) {
			CostInfo info = startMonitorOp(LimeConstants.OUT);
			ret = doOut(t);
			endMonitorOp(info, ret, true);
		} else {
			ret = doOut(t);
		}
       
		// Here to check replication policy.
		if (t != null && t instanceof ShareTuple)
			tryFR((ShareTuple) t);
		return t;
	}

	DracoTuple doOut(DracoTuple t) {
		t.setCurr(Location.HERE);
		if (t.getDest() == null || t.getDest().isInvalid()) {
			t.setDest(Location.HERE);
		}
		if (t.limeWrapper.id == null)
			t.limeWrapper.id = nextTID();

		if (t.getVID() == null)
			t.setVID(nextVID());

		return performUniOp(LimeConstants.OUT, t);
	}

	/**
	 * Out a Draco tuple using multicast directly
	 * 
	 * @param t the draco tuple that have been created
	 * @param wheretos the targets the <code>t</code> to be sent. If this is
	 * null, <code>t</code> will be sent to all the engaged hosts right now
	 * @return the out result tuple
	 * @author Zhang Jinxue
	 */
	public DracoTuple outToMulti(DracoTuple t, Location[] wheretos) {
		DracoTuple ret = null;
		if (t instanceof ShareTuple && ((ShareTuple) t).isReplica)
			throw new IllegalArgumentException("Cannot out a replica");

		if (enableMonitor) {
			CostInfo info = startMonitorOp(LimeConstants.OUTM);
			ret = doOutToMulti(t, wheretos);
			endMonitorOp(info, ret, true);
		} else {
			ret = doOutToMulti(t, wheretos);
		}
       
		// Here to check replication policy.
		if (t != null && t instanceof ShareTuple)
			tryFR((ShareTuple) t);
		return t;
	}

	/**
	 * Do the real multicast. The targets where to be sent is determined by
	 * <code>wheretos</code>, rather than the destination field of tuple
	 * itself.
	 * @param t the tuple to be sent
	 * @param wheretos the targets to be sent. If set null, <code>t</code>
	 * will be sent everywhere engaged.
	 * @return the tuple being sent
	 * @author Xing Jiankuan
	 */
	DracoTuple doOutToMulti(DracoTuple t, Location[] wheretos) {
		t.setCurr(Location.HERE);
		t.setDest(Location.UNSPECIFIED);
		if (t.limeWrapper.id == null)
			t.limeWrapper.id = nextTID();

		if (t.getVID() == null)
			t.setVID(nextVID());

		return performUniOpToMulti(LimeConstants.OUTM, t, wheretos);
	}
	
	/**
	 * 
	 * @param replica
	 * @return
	 * @author Xing Jiankuan
	 */
	DracoTuple doOutReplica(ShareTuple replica) {
		assert (replica.isReplica);

		replica.setCurr(Location.HERE);
		if (replica.getDest() == null || replica.getDest().isInvalid()) {
			replica.setDest(Location.HERE);
		}
		if (replica.limeWrapper.id == null)
			replica.limeWrapper.id = nextTID();

		if (replica.getVID() == null)
			replica.setVID(nextVID());

		return performUniOp(LimeConstants.OUT, replica);
	}

	/**
	 * 
	 * @param replicas
	 * @return
	 * @author Xing Jiankuan
	 */
	DracoTuple[] doOutgReplica(ShareTuple[] replicas) {
		for (ShareTuple replica : replicas) {
			assert (replica.isReplica);

			replica.setCurr(Location.HERE);
//			if (replica.getDest() == null || replica.getDest().isInvalid()) {
				replica.setDest(Location.HERE);
//			}
			if (replica.limeWrapper.id == null)
				replica.limeWrapper.id = nextTID();

			if (replica.getVID() == null)
				replica.setVID(nextVID());
//			System.out.println("%%%%%%%%%%% Add replica for " + replica.getVID());
		}
		return performMultiOp(LimeConstants.OUTG, replicas);
	}

	/**
	 * @param st
	 *            should be a Local Master
	 * @author Huang Xiaodong
	 */
	private void tryFR(ShareTuple st) {
		if (st.getDest().isHere() && st.isMaster()) {
			TupleType targetType = st.getType();

			Location[] wheretos = policyRegistrar.getFRHostsWithoutHere(targetType);
			if (wheretos.length == 0)
				return;
			
			ShareTuple replica = st.replicate(null);
			pushReplica(new ShareTuple[]{replica}, wheretos);
		}
	}
	
	/**
	 * Try to perform FR push
	 * @param dts
	 * @author Xing Jiankuan
	 */
	private void tryFR(ArrayList<ShareTuple> dts) {
		//Organize the tuples by their tuple types
		HashMap<TupleType, LinkedList<ShareTuple>> tuples = new HashMap<TupleType, LinkedList<ShareTuple>>();
		for (ShareTuple st : dts) {
			if (st.getDest().isHere() && st.isMaster()) {
				if (!tuples.containsKey(st.type))
					tuples.put(st.type, new LinkedList<ShareTuple>());
				tuples.get(st.type).add(st);
			}
		}
		
		//filter out tuples that are unnecessary to push
		ArrayList<ShareTuple> replicas = new ArrayList<ShareTuple>();
		for (TupleType type : tuples.keySet()) {
			Location[] wheretos = policyRegistrar.getFRHostsWithoutHere(type);
			if (wheretos.length == 0) {
				tuples.remove(type);
				continue;
			}

			//for those that need to push, push them
			List<ShareTuple> sts = tuples.get(type);
			replicas.clear();
			for (ShareTuple st : sts) {
				ShareTuple replica;
				replica = st.replicate(null);
				replicas.add(replica);
			}
			
			// convert array list to array
			ShareTuple[] replicaArr = new ShareTuple[replicas.size()];
			for (int i = 0; i < replicaArr.length; i++) {
				replicaArr[i] = (ShareTuple) replicas.get(i);
			}
			
			if (replicaArr.length == 0)
				return;
			
			pushReplica(replicaArr, wheretos);
		}
	}
	
	/**
	 * Out a group of DracoTuples directly
	 * 
	 * @param dts
	 *            the draco tuples that have been created
	 * @return the out result tuple
	 * @author Huang Xiaodong
	 */
	public DracoTuple[] outg(DracoTuple[] dts) {
		if (dts == null || dts.length < 0)
			throw new IllegalArgumentException(
					"the DracoTuples to out must be null or with length 0.");
		DracoTuple[] ret = null;
		if (enableMonitor) {
			CostInfo info = startMonitorOp(LimeConstants.OUTG);
			ret = doOutg(dts);
			endMonitorOp(info, ret, true);
		} else {
			ret = doOutg(dts);
		}
		// Here to check policy.
		ArrayList<ShareTuple> sts = new ArrayList<ShareTuple>();
		for(DracoTuple t: dts) {
			if(t instanceof ShareTuple) {
				sts.add((ShareTuple)t);
			}
		}
		
		tryFR(sts);
		return dts;
	}

	/**
	 * @param dts
	 * @param ret
	 * @author Xing Jiankuan
	 */
	DracoTuple[] doOutg(DracoTuple[] dts) {
		for (int i = 0; i < dts.length; ++i) {
			dts[i].setCurr(Location.HERE);
			if (dts[i].getDest() == null || dts[i].getDest().isInvalid()) {
				dts[i].setDest(Location.HERE);
			}
			if (dts[i].limeWrapper.id == null)
				dts[i].limeWrapper.id = nextTID();

			if (dts[i].getVID() == null)
				dts[i].setVID(nextVID());
		}

		return performMultiOp(LimeConstants.OUTG, dts);
	}

	/**
	 * Out a group of tuples to multiple destination, just as multicast. We make
	 * it public because it may be useful for application developer. <br>
	 * In implementation, we call <code>outg()</code> iterative for every
	 * LimeServer. For other information, refer to <code>outg</code>.<br>
	 * 
	 * @return an 2-dimension array of DracoTuple array instances that are sent
	 *         to all "wheretos". One destination, one result array in the
	 *         array.
	 */
	public DracoTuple[][] outgToMulti(DracoTuple dts[], Location[] wheretos) {
		if (dts == null || sagroup.util.Util.testNullArray(wheretos))
			throw new IllegalArgumentException("null value");

		DracoTuple[][] ret = new DracoTuple[wheretos.length][];
		for (int i = 0; i < wheretos.length; i++) {
			DracoTuple[] dups = new DracoTuple[dts.length];
			for (int j = 0; j < dts.length; j++) {
				dups[j] = dts[j].duplicate(wheretos[i]);
			}
			ret[i] = outg(dups);
		}
		return ret;
	}

	/**
	 * Out a group of tuples to all other engaged hosts.
	 * 
	 */
	public void outBroad(DracoTuple dt) {
		// parameter checking
		if (dt == null)
			throw new IllegalArgumentException("null value");
		
		if(LimeConstants.TREE_MULTICAST)
			outToMulti(dt, null);
		else {
			//send one-by-one
			LimeServerID[] hosts = getAllEngagedHosts();
			for (int i = 0; i < hosts.length; i++) {
				dt.setDest(new Location(hosts[i]));
				out(dt);
			}
		}
	}

	/**
	 * Out a group of tuples to all other engaged hosts.
	 * 
	 * @param tupleType
	 *            tuple's type
	 * @param t
	 *            tuples to be sent
	 * @return an array of TVRLimeTuple array instances that are sent to all
	 *         engaged hosts. One host, one result array in the array.
	 */
	public DracoTuple[][] outgBroad(DracoTuple[] dts) {
		// parameter checking
		if (dts == null)
			throw new IllegalArgumentException("null value");

		LimeServerID[] hosts = getAllEngagedHosts();
		Location[] wheretos = new Location[hosts.length];
		for (int i = 0; i < hosts.length; i++) {
			wheretos[i] = new Location(hosts[i]);
		}
		return outgToMulti(dts, wheretos);
	}

	/**
	 * Add replications.
	 * 
	 * @param replicaSrc
	 *            source location of replication
	 * @param entries
	 *            the replica request entries
	 * @author Huang Xiaodong
	 */
	public void addReplica(Location replicaSrc,
			AddReplicaRequestByEntryCommand.ReplicaRequestEntry[] entries) {
		out(new AddReplicaRequestByEntryCommand(entries, replicaSrc));
	}

	/**
	 * Remove replications.
	 * 
	 * @param masterLoc
	 *            Location of master
	 * @param vids
	 *            VersionLayerIDs of replications to remove.
	 * @author Huang Xiaodong
	 */
	public ShareTuple[] removeReplica(TupleID[] vids) {
		ShareTuple[] ret = new ShareTuple[vids.length];
		// PostingData helps to collect the vids with same
		// master location together
		CommandReactionListener.PostingData vidsToSend = new CommandReactionListener.PostingData();
		for (int i = 0; i < vids.length; ++i) {
			ret[i] = lrr.removeReplica(vids[i]); // remove local replicas

			if (ret[i] != null)
				vidsToSend.putData(ret[i].getMasterCurr(), vids[i]);
		}
		Location[] wheretos = vidsToSend.getAllDest();
		if (wheretos != null) {
			for (int i = 0; i < wheretos.length; ++i) {
				Object[] obj_vids = vidsToSend.getData(wheretos[i]);
				if (obj_vids != null) {
					TupleID[] tid_vids = new TupleID[obj_vids.length];
					for (int j = 0; j < obj_vids.length; ++j)
						tid_vids[j] = (TupleID) obj_vids[j];
					out(new RemoveReplicaRequestCommand(tid_vids, wheretos[i]));
				}
			}
		}
		return ret;
	}
	
	/**
	 * Push replicas to all the targets. All the <code>replicas</code>
	 * will be sent to all the targets.
	 * @param replicas replicas be sent
	 * @param targets the location where replicas are sent
	 * @author Xing Jiankuan
	 */
	public void pushReplica(ShareTuple[] replicas, Location[] targets) {
		if(LimeConstants.TREE_MULTICAST) {
			PushCommand pc = new PushCommand(replicas, Location.UNSPECIFIED); 
			outToMulti(pc, targets);
		} else {
			for(Location target: targets) {
				for(ShareTuple replica: replicas) {
					replica.setDest(target);
				}
				out(new PushCommand(replicas, target));
			}
		}
	}
	
	/**
	 * Push replicas to a single place, just directly send it
	 * @param replicas
	 * @param target
	 * @author Xing Jiankuan
	 */
	public void pushReplica(ShareTuple[] replicas, Location target) {
		out(new PushCommand(replicas, target));
	}

//	/**
//	 * Push Replicas to some targets by naive multi-unicast
//	 * 
//	 * @param replicas
//	 *            replications that need to push.
//	 * 
//	 * @author Huang Xiaodong
//	 */
//	public void pushReplicaByNaiveMulUnicast(ShareTuple[] replicas) {
//		if (replicas == null || replicas.length == 0) {
//			throw new IllegalArgumentException(
//					"Replications must not be null or with length 0.");
//		}
//		CommandReactionListener.PostingData tuplesToPush = new CommandReactionListener.PostingData();
//		for (int i = 0; i < replicas.length; ++i) {
//			replicaRegistrar.registerReplica(replicas[i].getVID(), replicas[i]
//					.getDest(), replicas[i].getVer());
//					
//			tuplesToPush.putData(replicas[i].getDest(), replicas[i]);
//		}
//		Location[] wheretos = tuplesToPush.getAllDest();
//		if (wheretos != null) {
//			for (int i = 0; i < wheretos.length; ++i) {
//
//				Object[] obj_data = tuplesToPush.getData(wheretos[i]);
//				if (obj_data != null) {
//					ShareTuple[] st_data = new ShareTuple[obj_data.length];
//					for (int j = 0; j < obj_data.length; ++j)
//						st_data[j] = (ShareTuple) obj_data[j];
//					
//					out(new PushCommand(st_data, wheretos[i]));
////					outToMulti(, wheretos);
//				}
//			}
//		}
//	}

	/**
	 * Deregister remote replications of local master with specified location,
	 * and send a ClearReplicaCommand to remote.
	 * 
	 * @param vids
	 * @param replicaLoc
	 * @author Huang Xiaodong
	 */
	public void clearReplica(TupleID[] vids, Location replicaLoc) {
		if (vids == null || vids.length == 0 || replicaLoc == null
				|| replicaLoc.equals(Location.UNSPECIFIED))
			throw new IllegalArgumentException(
					"Vids and replicaLoc must not be null or unspecified.");
		for (TupleID id : vids)
			replicaRegistrar.deregisterReplica(id, replicaLoc);
		out(new ClearReplicaCommand(vids, replicaLoc));
	}

	/**
	 * Get all engaged hosts' lime server IDs (except self).
	 * 
	 * @Request: nothing.
	 * @Modifies: nothing.
	 * @Effects: If no Host, then return zero length array. else return array of
	 *           engaged LimeServers' IDs.
	 * @return an array of hosts' lime server IDs.
	 */
	public LimeServerID[] getAllEngagedHosts() {
		LimeTuple[] ts = LimeSystemTupleSpace
				.get()
				.rdg(
						new Tuple()
								.add(
										new Field()
												.setValue(LimeSystemTupleSpace.HOST))
								.add(
										new Field()
												.setValue(lights.dracots.MatchAllField.MULTI_FIELD_WILDCARD)));
		if (ts == null) {
			return new LimeServerID[0];
		}
		LimeServerID[] toReturn = new LimeServerID[ts.length];
		for (int i = 0; i < ts.length; i++) {
			toReturn[i] = (LimeServerID) ((IValuedField) ts[i].getTuple()
					.get(1)).getValue();
		}
		return toReturn;
	}

	public Location[] getAllEngagedLocations() {
		LimeServerID[] ids = this.getAllEngagedHosts();
		Location[] ls = new Location[ids.length];
		int i = 0;
		for (LimeServerID id : ids) {
			ls[i] = new Location(id);
			i++;
		}
		return ls;
	}

	/**
	 * Add a series of strong reactions
	 * 
	 * @param versionedReactions
	 *            array of reactions to be added
	 * @return the added reactions' IDs
	 * @author Xing Jiankuan
	 */
	public ReactionID[] addStrongReaction(DracoReaction[] reactions) {
		// parameter checking
		if (Util.testNullArray(reactions))
			throw new IllegalArgumentException("null value");

		for (int i = 0; i < reactions.length; i++)
			if (!reactions[i].getTemplate().getCurr().isHere())
				throw new InvalidLocationException(
						"strong reactions cannot be remote or ubiquitous");

		// end of check
		Result result = new Result();
		Reaction[] rs = DracoReaction.getLimeReactions(reactions);
		ReactionOp ro = new ReactionOp(LimeConstants.INSTALL_STRONG, rs, result);
		lop.postOperation(ro);
		return (ReactionID[]) result.get();
	}

	/**
	 * Removes strong reactions from this host. <br>
	 * If the ReactionIDs objects do not correspond to strong reactions
	 * installed previously on this host, the reactions will not be
	 * deregistered. You can uninstall a strong reaction from a remote host
	 * using an agent (.. you still need the ReactionID "handle").
	 * 
	 * @param ids
	 *            of the (registered) reactions to remove
	 * @return boolean[], where <code>boolean[i]</code> is <code>true</code> if
	 *         reaction <code>i</code> was successfully removed,
	 *         <code>false</code> otherwise
	 */
	public boolean[] removeStrongReaction(ReactionID[] ids) {
		// parameter checking
		if (Util.testNullArray(ids))
			throw new IllegalArgumentException("null value");

		Result result = new Result();
		ReactionOp ro = new ReactionOp(LimeConstants.REMOVE_STRONG, ids, result);
		lop.postOperation(ro);
		return (boolean[]) result.get();
	}

	/**
	 * Add a series of weak reactions
	 * 
	 * @param versionedReactions
	 *            array of reactions to be added
	 * @return the added reactions' IDs
	 * @author Xing Jiankuan
	 */
	public ReactionID[] addWeakReaction(DracoReaction[] reactions) {
		// parameter checking
		if (Util.testNullArray(reactions))
			throw new IllegalArgumentException("null value");

		Result result = new Result();
		Reaction[] rs = DracoReaction.getLimeReactions(reactions);
		ReactionOp ro = new ReactionOp(LimeConstants.INSTALL_WEAK, rs, result);
		lop.postOperation(ro);
		return (ReactionID[]) result.get();
	}

	public boolean[] removeWeakReaction(ReactionID[] ids) {
		// parameter checking
		if (Util.testNullArray(ids))
			throw new IllegalArgumentException("null value");

		Result result = new Result();
		ReactionOp ro = new ReactionOp(LimeConstants.REMOVE_WEAK, ids, result);
		lop.postOperation(ro);
		boolean[] ret = (boolean[]) result.get();
		for (int i = 0; i < ret.length; i++) {
			if (!ret[i]) {
				LOG.warning("fail remove weak reaction with id: " + ids[i]);

			}
		}
		return ret;
	}

	/**
	 * Output the operation cost info log to an excel file, and choose whether
	 * clear the logCache or not after writing log
	 * 
	 * @param outputPath
	 *            of log, if null, use the default path "./log/"
	 * @param clearCache
	 *            to clear logCache in memory or not
	 * @return the full outputPath including file name
	 * @author Huang Xiaodong
	 */
	public String outputLog(String outputPath, boolean clearCache) {
		String fullPath = monitor.writeLog(outputPath);
		if (clearCache)
			monitor.clearLog();
		return fullPath;
	}

	public TupleID[] getUnusedReplicaByType(TupleType type) {
		return lrr.getUnusedReplicaByType(type);
	}

//	void updateReplica(ShareTuple oldTuple, ShareTuple newTuple) {
//		synchronized (this) {
//
//			try {
//				Result result = new Result();
//				TupleOp to = new TupleOp(LimeConstants.UPDATE_REPLICA,
//						new ITuple[] {
//								oldTuple.toLimeWrapper().limifyWrapper(),
//								newTuple.toLimeWrapper().limifyWrapper() },
//						result);
//				lop.postOperation(to);
//				// if (ret.getVID().equals(newTuple.getVID()))
//				// dmaker.insertUpdateRecord(newTuple, 1);
//			} catch (Exception ex) {
//
//			}
//		}
//	}
	void updateReplica(ShareTuple oldTuple, ShareTuple newTuple) {
		
		lrr.rupdate(oldTuple,newTuple);
	}
	/**
	 * Perform out, in, inp, rd, rdp
	 * 
	 * @param op
	 *            operation type
	 * @param t
	 *            operation's tuple or template
	 * @return the result tuple
	 * @author Xing Jiankuan
	 */
	private DracoTuple performUniOp(byte op, DracoTuple t) {
		if (t == null)
			throw new NullPointerException();

		// construct operation
		Result result = new Result();
		TupleOp to = new TupleOp(op, new ITuple[] { t.toLimeWrapper()
				.limifyWrapper() }, result);
		lop.postOperation(to);

		// return the result
		try {
			return DracoTuple.create((ITuple) result.get());
		} catch (ClassNotFoundException e) {
			throw new RuntimeException(e);
		}
	}

	/**
	 * Perform outm
	 * 
	 * @param op
	 *            operation type
	 * @param t
	 *            operation's tuple or template
	 * @return the result tuple
	 * @author Xing Jiankuan
	 */
	private DracoTuple performUniOpToMulti(byte op, DracoTuple t, Location[] wheretos) {
		if (t == null)
			throw new NullPointerException();

		// construct operation
		Result result = new Result();
		TupleOp to = new TupleOp(op, new ITuple[] { t.toLimeWrapper()
				.limifyWrapper() }, wheretos, result);
		lop.postOperation(to);

		// return the result
		try {
			return DracoTuple.create((ITuple) result.get());
		} catch (ClassNotFoundException e) {
			throw new RuntimeException(e);
		}
	}

	
	/**
	 * Perform ing, rdg
	 * 
	 * @param op
	 *            operation type
	 * @param t
	 *            operation's template
	 * @return the result tuple array
	 * @author Xing Jiankuan
	 */
	private DracoTuple[] performMultiOp(byte op, DracoTuple t) {

		if (t == null)
			throw new NullPointerException();

		// construct operation
		Result result = new Result();
		TupleOp to = new TupleOp(op, new ITuple[] { t.toLimeWrapper()
				.limifyWrapper() }, result);
		lop.postOperation(to);

		// get the result
		ITuple[] tuples = (ITuple[]) result.get();
		if (tuples == null)
			return DracoConstants.EMPTY_DRACOTUPLE_ARR;

		// convert ITuple array to DracoTuple array
		try {
			DracoTuple[] ret = Util.convertITupleArrayToDracoTupleArray(tuples);
			return ret;
		} catch (ClassNotFoundException e) {
			throw new RuntimeException(e);
		}
	}

	/**
	 * Perform outg
	 * 
	 * @param op
	 *            operation type
	 * @param t
	 *            operation's tuples
	 * @return the result tuple array
	 * @author Xing Jiankuan
	 */
	private DracoTuple[] performMultiOp(byte op, DracoTuple[] dts) {

		if (dts == null)
			throw new NullPointerException();

		// construct operation
		Result result = new Result();
		ITuple[] tuples = Util.convertDracoTupleArrayToITupleArray(dts);
		TupleOp to = new TupleOp(op, tuples, result);
		lop.postOperation(to);

		return dts;
	}

	PolicyDecisionMaker getPolicyDecisionMaker() {
		return dmaker;
	}

	LocalReplicaRegistrar getLocalReplicaRegistrar() {
		return lrr;
	}

	DracoTuple doUpdateRdp(ShareTuple template) {
		DracoTuple ret = null;
		// try to read local replica if template indicate remote host or
		// unspecified
		// if it is FR, always replica can be found
		if (!template.getCurr().equals(Location.HERE)
				&& template instanceof ShareTuple) {

			ShareTuple stemplate = (ShareTuple) template
					.duplicateTemplate(template.getDest());

			stemplate.setCurr(Location.HERE);
			stemplate.setDest(Location.HERE);
			stemplate.setReplica(true);
			ret = performUniOp(LimeConstants.RDP, stemplate);
			if (ret != null) // found

				return ret;

		}
		// perform the conventional rdp
		ret = performUniOp(LimeConstants.RDP, template);

		return ret;

	}

	class TryRC implements Runnable {
		ShareTuple st;

		TryRC(ShareTuple st) {
			this.st = st;
		}

		@Override
		public void run() {

			if (st.isReplica)
				lrr.touch(st);
			else
				tryRC(st);
		}
	}

	class TryRCs implements Runnable {
		DracoTuple[] st;

		TryRCs(DracoTuple[] st) {
			this.st = st;
		}

		@Override
		public void run() {
			// Here to check the policy.
			if (st != null)
				tryRCs(st);

			for (DracoTuple dt : st) {
				if (dt instanceof ShareTuple) {
					ShareTuple st = (ShareTuple) dt;
					if (dt.isLocal() && st.isReplica())
						lrr.touch(st);
				}
			}
		}
	}
}

